package com.klun.hdfs.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Value;

import java.io.IOException;
import java.util.ArrayList;


public class HbaseConfig {
    public static Configuration config;

    static {
        // 使用Hbase的单例方法实例化
        config = HBaseConfiguration.create();
        config.set("hbase.zookeeper.quorum", "");
        config.set("hbase.zookeeper.property.clientPort", "2181");
    }

    // 判断表是否存在
    public static boolean isTableExist(String tableName) throws IOException {
        // 在 hbase 中管理，访问表先创建 hbaseadmin
        //Connection connection = ConnectionFactory.createConnection(config);
        //HBaseAdmin admin = (HBaseAdmin)connection.getAdmin();
        HBaseAdmin admin = new HBaseAdmin(config);
        return admin.tableExists(tableName);
    }

    // 创建表
    public static void createTable(String tableName, String... columnFamily) throws IOException {
        HBaseAdmin admin = new HBaseAdmin(config);
        // 判断表是否存在
        if (isTableExist(tableName)) {
            System.out.println("表已经存在");
        } else {
            // 创建表属性对象，表名需要转字节
            HTableDescriptor descriptor = new HTableDescriptor(TableName.valueOf(tableName));
            // 创建多个列族
            for (String cf : columnFamily) {
                descriptor.addFamily(new HColumnDescriptor(cf));
            }
            // 根据对表的配置，创建表
            admin.createTable(descriptor);
            System.out.println(" create table success");
        }
    }

    // 删除表
    public static void dropTable(String tableName) throws IOException {
        HBaseAdmin admin = new HBaseAdmin(config);
        if (isTableExist(tableName)) {
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
            System.out.println("success drop table");
        } else {
            System.out.println("fail drop table");
        }
    }

    // 向表中插入数据
    public static void addRowData(String tableName, String rowKey, String columnFamily, String column, String value) throws IOException {
        // 创建 HTable 对象
        HTable htable = new HTable(config, tableName);
        // 向表中插入数据
        Put put = new Put(Bytes.toBytes(rowKey));
        // 向put对象中组装数据
        put.add(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
        htable.put(put);
        htable.close();
        System.out.println("add data success");
    }

    // 删除多行数据
    public static void deleteMultiRow(String tableName, String... rows) throws IOException {
        HTable hTable = new HTable(config, tableName);
        ArrayList<Delete> deletes = new ArrayList<>();
        for (String row : rows) {
            Delete delete = new Delete(Bytes.toBytes(row));
            deletes.add(delete);
        }
        hTable.delete(deletes);
        hTable.close();
    }

    // 获取所有数据
    public static void getAllRows(String tableName) throws IOException {
        HTable hTable = new HTable(config, tableName);
        // 得到用于扫描 region 的对象
        Scan scan = new Scan();
        // 使用 HTable 得到 resultScanner 实现类的对象
        ResultScanner scanner = hTable.getScanner(scan);
        for (Result result : scanner) {
            Cell[] cells = result.rawCells();
            for (Cell cell : cells) {
                // 得到rowKey
                System.out.println("行键："+Bytes.toString(CellUtil.cloneRow(cell)));
                // 得到列族
                System.out.println("列族"+Bytes.toString(CellUtil.cloneFamily(cell)));
                System.out.println("列"+Bytes.toString(CellUtil.cloneQualifier(cell)));
                System.out.println("值"+Bytes.toString(CellUtil.cloneValue(cell)));
            }
        }
    }

    // 获取某一行数据
    public static void getRow(String tableName,String rowKey) throws IOException{
        HTable hTable = new HTable(config, tableName);
        Get get = new Get(Bytes.toBytes(rowKey));
        get.setMaxVersions(); //显示所有版本
        get.setTimeStamp(1L); //指定时间戳的版本
        Result result = hTable.get(get);
        for (Cell cell : result.rawCells()) {
            System.out.println("行键："+Bytes.toString(result.getRow()));
            System.out.println("列族"+Bytes.toString(CellUtil.cloneFamily(cell)));
            System.out.println("列"+Bytes.toString(CellUtil.cloneQualifier(cell)));
            System.out.println("值"+Bytes.toString(CellUtil.cloneValue(cell)));
            System.out.println("时间戳："+cell.getTimestamp());
        }
    }
    // 获取某一行指定“列族：列”的数据
    public static void getRowQualifier(String tableName,String rowKey,String family,String qualifier) throws IOException {
        HTable hTable = new HTable(config, tableName);
        Get get = new Get(Bytes.toBytes(rowKey));

        get.addColumn(Bytes.toBytes(family),Bytes.toBytes(qualifier));

        Result result = hTable.get(get);
        for (Cell cell : result.rawCells()) {
            System.out.println("行键："+Bytes.toString(result.getRow()));
            System.out.println("列族"+Bytes.toString(CellUtil.cloneFamily(cell)));
            System.out.println("列"+Bytes.toString(CellUtil.cloneQualifier(cell)));
            System.out.println("值"+Bytes.toString(CellUtil.cloneValue(cell)));
            System.out.println("时间戳："+cell.getTimestamp());
        }

    }
}
